package com.gitee.huanminabc.utils_tools.redis.redisTemplate.utils;

import com.gitee.huanminabc.utils_tools.spring_base.utils.SpringContextUtil;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * @Description  将时间段内任务，合并成1次操作来处理
 * 可以说是一个方便的延迟队列, 但是不一样的是,这个延迟队列是将这个时间段内的任务合并,然后在下一个时间段一起处理 , 这是mq无法做到的
 * 误差在1秒左右
 * 注意: 如果数据量过大,会导致redis内存占用过大, 需要根据实际情况使用, 比如突然来了10万条数据,但是我们在一个时间段内只处理100条,那么剩余的数据就会一直在redis里面
 * 本来物理机的内存是不大的一般都是32或者16g
 *
 * @Author huanmin
 * @Date 2024/4/23 上午10:47
 */
//@TODO: 有时间可以参考MergeDelayTimeTask改造下
@Slf4j
public class RedisMergeDelayTimeTask<T>  {
    /**
     * 将多长时间的多次操作合并成1次，单位：秒
     */
    private final  int intervalSecond;
    /**
     * 每次处理多少条数据
     */
    private  final int perBatchCount;
    private final  String LIST_KEY;
    private final  String LOCK_KEY;
    private  final Consumer<List<T>> batchBizMethod;
    //因为在这个逻辑中是按照时间段来处理的,没有啥并发,基本是窜行的,所以这里设置3个线程就够了
    private final ScheduledExecutorService scheduledThreadPool ;
    private final AtomicBoolean triggerScheduleNeeded = new AtomicBoolean(true);
    private  static   RedisTemplateUtil redisTemplateUtil;
    private  static RedisDistributeLock redisDistributeLock;

    static {
        Timer timer = new Timer();
        //持续检查是否为空, 一般spring容器启动不会太快这里做了一个延迟检查,检查到了后就取消定时任务
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                if (redisTemplateUtil == null) {
                    redisTemplateUtil = SpringContextUtil.getBean(RedisTemplateUtil.class);
                }
                if (redisDistributeLock == null) {
                    redisDistributeLock = SpringContextUtil.getBean(RedisDistributeLock.class);
                }
                //如果redisTemplateUtil和redisDistributeLock都不为空,则取消定时任务
                if (redisTemplateUtil != null && redisDistributeLock != null) {
                    timer.cancel();
                    synchronized (RedisMergeDelayTimeTask.class){
                        RedisMergeDelayTimeTask.class.notifyAll();
                    }
                }
            }
        }, 500,  1000);
    }



    /**
     * 初始化
     * @param intervalSecond 时间间隔
     * @param perBatchCount 每次处理多少条数据, 前提是在这个时间段内有这么多数据,如果没有这么多数据,则取出所有数据,如果有超过这么多数据,则在下一个时间段继续处理
     * @param tClass 当前类为了分布式锁的key
     * @param batchBizMethod 批次处理逻辑代码
     */
    public RedisMergeDelayTimeTask(String taskName,int intervalSecond, int perBatchCount, Class tClass, Consumer<List<T>> batchBizMethod){

        //intervalSecond最低为1秒
        if (intervalSecond < 1) {
            intervalSecond = 1;
        }
        //perBatchCount最低为1
        if (perBatchCount < 1) {
            perBatchCount = 1;
        }
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat(taskName + "-%d").build();
        this.scheduledThreadPool = Executors.newScheduledThreadPool(3, namedThreadFactory);
        this.intervalSecond = intervalSecond;
        this.perBatchCount = perBatchCount;
        this.batchBizMethod = batchBizMethod;
        this.LIST_KEY = "MergeTimeTaskList" + tClass;
        this.LOCK_KEY = "MergeTimeTaskLock" + tClass; //用于多台服务器时，保证只有一台服务器执行
    }


    //生产者
    public void produce(T t){
        check();
        //将任务放入队列
        redisTemplateUtil.lSet(this.LIST_KEY,t, TimeUnit.HOURS.toMillis(1));
        //判断是否需要触发新任务
        if (triggerScheduleNeeded.compareAndSet(true, false)) {
            log.info("MergeTimeTask produce");
            //启动一个新任务延迟intervalSecond秒后执行
            scheduledThreadPool.schedule(this::consumeMsg, intervalSecond, TimeUnit.SECONDS);
        }
    }



    private void consumeMsg() {
        //设置为true，保证永远都有新任务去处理数据
        triggerScheduleNeeded.set(true);
        //加锁防止多个服务器同时执行, 没抢到的直接丢弃, 因为抢到的会执行
        //这会有一个问题,如果执行的代码太快了,锁被释放了,那么就导致会出现创建2个定时任务了,我们需要将锁不释放和延迟时间设置成一样-1
        //所以采用了executeNoUnlock方法
        redisDistributeLock.executeNoUnlock(LOCK_KEY, intervalSecond , () -> {
            //获取队列长度
            long size = redisTemplateUtil.lGetListSize(this.LIST_KEY);
            if (size == 0) {
                //消费--发现无任何数据可供消费
                return ;
            }
            //获取队列中这个时间段内需要处理的数据
            List<Object> list = redisTemplateUtil.lGet(this.LIST_KEY, 0, perBatchCount - 1); //每次取出perBatchCount条数据限制处理的数据量
            //说明还有数据未处理，移交到下一个时间段继续触发新任务,  如果这里不这样做,那么就需要等待下次produce方法调用才能继续消费
            long newSize = redisTemplateUtil.lGetListSize(this.LIST_KEY);
            if (newSize>0) {
                scheduledThreadPool.schedule(this::consumeMsg, intervalSecond, TimeUnit.SECONDS);
            }
            //消费
            try {
                batchBizMethod.accept((List<T>) list);
            } catch (Exception e) {
                log.error("MergeTimeTask consumeMsg error",e);
            }
        });

    }

    //如果没有初始化,则等待初始化
    private void   check(){
        if (redisTemplateUtil == null || redisDistributeLock == null) {
            synchronized (RedisMergeDelayTimeTask.class){
                try {
                    RedisMergeDelayTimeTask.class.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            }

        }
    }

}
